package org.example.springbatchdemo.batch;

import org.example.springbatchdemo.batch.model.Person;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;

import javax.sql.DataSource;

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public JdbcCursorItemReader<Person> reader(@Qualifier("sourceDataSource") DataSource sourceDataSource) {
        return new JdbcCursorItemReaderBuilder<Person>()
            .name("personReader")
            .dataSource(sourceDataSource)
            .sql("SELECT id, name, age, created_at FROM users")
            .rowMapper((rs, rowNum) -> {
                Person person = new Person();
                person.setId(rs.getLong("id"));
                person.setName(rs.getString("name"));
                person.setAge(rs.getInt("age"));
                person.setCreatedAt(rs.getTimestamp("created_at"));
                return person;
            })
            .fetchSize(1000)
            .build();
    }

    @Bean
    public JdbcBatchItemWriter<Person> writer(@Qualifier("targetDataSource") DataSource targetDataSource) {
        return new JdbcBatchItemWriterBuilder<Person>()
            .dataSource(targetDataSource)
            .sql("INSERT INTO users1 (id, name, age, created_at) VALUES (?, ?, ?, ?)")
            .itemPreparedStatementSetter((person, ps) -> {
                ps.setLong(1, person.getId());
                ps.setString(2, person.getName());
                ps.setInt(3, person.getAge());
                ps.setTimestamp(4, person.getCreatedAt());
            })
            .build();
    }

    @Bean
    public Step migrationStep(@Qualifier("sourceTransactionManager") DataSourceTransactionManager transactionManager,
        JdbcCursorItemReader<Person> reader,
        JdbcBatchItemWriter<Person> writer) {
        return stepBuilderFactory.get("migrationStep")
            .<Person, Person>chunk(1000)
            .reader(reader)
            .writer(writer)
            .transactionManager(transactionManager)
            .build();
    }

    @Bean
    public Job migrationJob(Step migrationStep) {
        return jobBuilderFactory.get("migrationJob")
            .start(migrationStep)
            .build();
    }


}